/**
 * @Author: Routz
 * @since: 2020/10/27 23:42
 */
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.Arrays;

/**
 * A simple class that monitors the data and existence of a ZooKeeper
 * node. It uses asynchronous ZooKeeper APIs.
 */
public class DataMonitor implements Watcher, StatCallback {

    ZooKeeper zk;
    String znode;
    Watcher chainedWatcher;
    boolean dead;
    DataMonitorListener listener;
    byte prevData[];

    /**
     * The DataMonitor class has the meat of the ZooKeeper logic. It is mostly asynchronous and event driven
     * @param zk
     * @param znode
     * @param chainedWatcher
     * @param listener
     */
    public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,
                       DataMonitorListener listener) {
        this.zk = zk;
        this.znode = znode;
        this.chainedWatcher = chainedWatcher;
        this.listener = listener;
        // Get things started by checking if the node exists. We are going
        // to be completely event driven
        // The call to ZooKeeper.exists() checks for the existence of the znode,
        //  sets a watch, and passes a reference to itself (this) as the completion callback object.
        // In this sense, it kicks things off, since the real processing happens when the watch is triggered.
        // Note
        //Don't confuse the completion callback with the watch callback.
        // The ZooKeeper.exists() completion callback,
        // which happens to be the method StatCallback.processResult() implemented in the DataMonitor object,
        // is invoked when the asynchronous setting of the watch operation (by ZooKeeper.exists()) completes on the server.
        //
        //The triggering of the watch, on the other hand, sends an event to the Executor object,
        // since the Executor registered as the Watcher of the ZooKeeper object.
        //
        //As an aside, you might note that the DataMonitor could also register itself as the Watcher
        // for this particular watch event.
        // This is new to ZooKeeper 3.0.0 (the support of multiple Watchers).
        // In this example, however, DataMonitor does not register as the Watcher.
        zk.exists(znode, true, this, null);
    }

    /**
     * Other classes use the DataMonitor by implementing this method
     * The DataMonitor object uses it to communicate back to its container, which is also the Executor object.
     */
    public interface DataMonitorListener {
        /**
         * The existence status of the node has changed.
         * the Executor decides whether to start up or shut down per the requirements.
         * Recall that the requires say to kill the executable when the znode ceases to exist.
         */
        void exists(byte data[]);

        /**
         * The ZooKeeper session is no longer valid.
         * the Executor decides whether or not to shut itself down in response to the ZooKeeper connection permanently disappearing.
         * @param rc the ZooKeeper reason code
         */
        void closing(int rc);
    }

    /**
     * 监控事件
     * DataMonitor processes watch events
     * If the client-side ZooKeeper libraries can re-establish the communication channel (SyncConnected event)
     *  to ZooKeeper before session expiration (Expired event)
     *  all of the session's watches will automatically be re-established with the server
     *  (auto-reset of watches is new in ZooKeeper 3.0.0).
     *  See ZooKeeper Watches in the programmer guide for more on this.
     *  A bit lower down in this function, when DataMonitor gets an event for a znode,
     *  it calls ZooKeeper.exists() to find out what has changed.
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {
        String path = event.getPath();
        if (event.getType() == Event.EventType.None) {
            // We are are being told that the state of the
            // connection has changed
            switch (event.getState()) {
                case SyncConnected:
                    // In this particular example we don't need to do anything
                    // here - watches are automatically re-registered with
                    // server and any watches triggered while the client was
                    // disconnected will be delivered (in order of course)
                    System.out.println("SyncConnected");
                    break;
                case Expired:
                    // It's all over
                    dead = true;
                    listener.closing(KeeperException.Code.SessionExpired);
                    break;
            }
        } else {
            if (path != null && path.equals(znode)) {
                // Something has changed on the node, let's find out
                zk.exists(znode, true, this, null);
            }
        }
        if (chainedWatcher != null) {
            chainedWatcher.process(event);
        }
    }

    /**
     * completion callback: zookeeper.exists()完成后执行
     * 1. checks the error codes for znode existence, fatal errors, and recoverable errors.
     * 2. If the file (or znode) exists, it gets the data from the znode,
     * 3. invoke the exists() callback of Executor if the state has changed.
     * Note, it doesn't have to do any Exception processing for the getData call
     *  because it has watches pending for anything that could cause an error:
     *  if the node is deleted before it calls ZooKeeper.getData(),
     *  the watch event set by the ZooKeeper.exists() triggers a callback;
     *  if there is a communication error, a connection watch event fires when the connection comes back up.
     *
     * @param rc
     * @param path
     * @param ctx
     * @param stat
     */
    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        boolean exists;
        switch (rc) {
            case Code.Ok:
                exists = true;
                break;
            case Code.NoNode:
                exists = false;
                break;
            case Code.SessionExpired:
            case Code.NoAuth:
                dead = true;
                listener.closing(rc);
                return;
            default:
                // Retry errors
                zk.exists(znode, true, this, null);
                return;
        }

        byte b[] = null;
        if (exists) {
            try {
                b = zk.getData(znode, false, null);
            } catch (KeeperException e) {
                // We don't need to worry about recovering now. The watch
                // callbacks will kick off any exception handling
                e.printStackTrace();
            } catch (InterruptedException e) {
                return;
            }
        }
        if ((b == null && b != prevData)
                || (b != null && !Arrays.equals(prevData, b))) {
            listener.exists(b);
            prevData = b;
        }
    }
}
